Initial implementation of Pipeline DLQ by kkondaka · Pull Request #5277 · opensearch-project/data-prepper

kkondaka · 2024-12-20T17:20:36Z

Description

[Describe what this change achieves]

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
[X ] Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

dlvenable · 2025-06-25T16:39:41Z

+        output(records, null);
+    }
+
+    void output(Collection<T> records, PipelineIf failurePipeline);


Should we have a setFailurePipeline method instead? I suspect the sinks are going to need to hang on this to in order to use it later in the code.

Also, rather than passing a PipelineIf, I'd probably pass this to the sinks.

interface FailurePipeline { void writeAll(Collection<T> records); }

All the sinks care about is writing to the pipeline. I don't think they care about the Source object itself. Also, the Source interface doesn't support writing.

I'd even think that the implementation of FailurePipeline would write directly to the Buffer.

@dlvenable Sinks that want to hang on can set that themselves, right? Why introduce a new API when the existing API can serve the purpose?

dlvenable · 2025-06-25T16:42:09Z

+
+import org.opensearch.dataprepper.model.source.Source;
+
+public interface PipelineIf {


What does PipelineIf mean?

@dlvenable I guess I meant it to be "Pipeline Interface" as a pointer to pipeline from Source. I am open to suggestions on the naming

I needed an interface in data-prepper-api directory so that I can use it in other interfaces in the directory

dlvenable · 2025-06-25T16:45:42Z

 public class DataPrepperConfiguration implements ExtensionsConfiguration, EventConfigurationContainer {
    static final Duration DEFAULT_SHUTDOWN_DURATION = Duration.ofSeconds(30L);

+    static final String DEFAULT_FAILURE_PIPELINE_NAME = "dlq";


I think we should give this a different name.

default_failure_pipeline

@dlvenable Raj prefers the well known name like dlq. My idea was to provide a way to change this name in a future PR.

I think the name dlq already implies something since our DLQs are different from any other concept. So I think a failure pipeline is clearer. Maybe default_dlq_pipeline instead?

or maybe just dlq_pipeline?

dlvenable · 2025-06-25T16:46:21Z

+    }
+
+    @Override
+    public void sendFailedEvents(Collection<Record<Event>> records) {


This is good.

dlvenable · 2025-07-14T17:12:52Z

+     * @return FailurePipeline returns failure pipeline
+     * @since 2.12
+     */
+    default FailurePipeline getFailurePipeline() {


I don't think we need these in the interface. Each Buffer can handle setFailurePipeline as needed.

dlvenable · 2025-07-14T17:17:26Z

+     * @return FailurePipeline returns failure pipeline
+     * @since 2.12
+     */
+    default FailurePipeline getFailurePipeline() {


You don't need the getter on the interface.

dlvenable · 2025-07-14T17:56:32Z

+     * @return FailurePipeline returns failure pipeline
+     * @since 2.12
+     */
+    default FailurePipeline getFailurePipeline() {


You don't need this on the interface.

dlvenable · 2025-07-14T17:56:45Z

+     * @return FailurePipeline returns failure pipeline
+     * @since 2.12
+     */
+    default FailurePipeline getFailurePipeline() {


You don't need this on the interface.

dlvenable · 2025-07-14T17:59:51Z

 public class DataPrepperConfiguration implements ExtensionsConfiguration, EventConfigurationContainer {
    static final Duration DEFAULT_SHUTDOWN_DURATION = Duration.ofSeconds(30L);

+    static final String DEFAULT_FAILURE_PIPELINE_NAME = "dlq";


I think the name dlq already implies something since our DLQs are different from any other concept. So I think a failure pipeline is clearer. Maybe default_dlq_pipeline instead?

graytaylor0 · 2025-07-28T22:22:51Z

+        try {
+            buffer.writeAll(records, DEFAULT_WRITE_TIMEOUT);
+        } catch (Exception e) {
+            LOG.error("Failed to write to failure pipeline");


Will we hit this if failure pipeline buffer ever gets full?

yes, I guess I could add some retries here. But overall, we can't wait here forever.

graytaylor0 · 2025-07-28T22:24:02Z

@@ -107,7 +107,9 @@ Collection runProcessorsAndProcessAcknowledgements(List<Processor> processors, C
                }
            } catch (final Exception e) {
                LOG.error("A processor threw an exception. This batch of Events will be dropped, and their EventHandles will be released: ", e);


Maybe we change this to log that it's going to failure pipeline if it's enabled and only log the dropped message when failure pipeline doesn't exist?

kkondaka · 2025-08-04T05:07:39Z

I am also thinking that I should rename "FailurePipeline" to "NoSourcePipeline". We may be using this pipeline in other cases as well. @dlvenable what do you think?

Signed-off-by: Kondaka <krishkdk@amazon.com>

graytaylor0 · 2025-08-11T16:32:13Z

            @JsonProperty("sink") final List<SinkModel> sinks,
            @JsonProperty("workers") final Integer workers,
            @JsonProperty("delay") final Integer delay) {
-        checkArgument(Objects.nonNull(source), "Source must not be null");


Why is source not required anymore? Doesn't even DLQ pipeline have a pipeline source?

In the PIpelineModel, it does not.

@kkondaka , Do we have validations elsewhere on this?

@graytaylor0 DLQ pipelines do not have source because source/processor/buffer/sink can end events to DLQ pipeline. So, the source is the new "HeadlessPipelineSource" that I added. This source is created automatically for a DLQ pipeline. It is not configurable.

@dlvenable. Yes, there are validations that fail if a source is not specified.

graytaylor0 · 2025-08-11T16:35:12Z

+            for (Map.Entry<String, Pipeline> pipelineEntry : pipelineMap.entrySet()) {
+                if (!(pipelineEntry.getKey().equals(failurePipelineName))) {
+                    pipelineEntry.getValue().setFailurePipeline(failurePipeline);
+                    acknowledgementsEnabled = acknowledgementsEnabled || pipelineEntry.getValue().areAcknowledgementsEnabled();


Why not just do

acknowledgementsEnabled = pipelineEntry.getValue().areAcknowledgementsEnabled();

If there are 10 sub pipelines, and first 9 has ack enabled and the 10th doesn't then the final result would be false! (In fact, this shouldn't happen but just want to be sure) and we want to "release events" when acks are enabled.

graytaylor0 · 2025-08-11T16:36:34Z

+                numberOfEventsSuccessful.increment(records.size());
+                break;
+            } catch (Exception e) {
+                LOG.error(NOISY, "Failed to write to failure pipeline");


We should log the exception message here.

graytaylor0 · 2025-08-11T16:38:42Z

-                LOG.error("A processor threw an exception. This batch of Events will be dropped, and their EventHandles will be released: ", e);
-                if (inputEvents != null) {
+                if (pipeline.getFailurePipeline() != null) {
+                    pipeline.getFailurePipeline().sendEvents(records);


We may still want a log here

LOG.error("A processor threw an exception. This batch of Events will be sent to the pipeline DLQ, and their EventHandles will be released: ", e);

@graytaylor0 LOG.error() automatically logs the exception, right?

dlvenable · 2025-08-11T17:26:12Z

+    }
+
+    @Test
+    //@Timeout(value = 2000, unit = TimeUnit.MILLISECONDS)


Please remove this comment.

dlvenable · 2025-08-11T17:26:25Z

+        Collection<Record<Event>> records = mock(Collection.class);
+        failurePipeline.sendEvents(records);
+        verify(headlessPipelineSource).sendEvents(records);
+        //assertThat(testPipeline.areAcknowledgementsEnabled(), equalTo(false));


Please remove.

dlvenable · 2025-08-11T17:26:40Z

+        processorSets.forEach(processorSet -> processorSet.forEach(processor -> {
+        assertThat(((TestProcessor)processor).getFailurePipeline(), equalTo(failurePipeline));
+        }));
+        for (Sink sink: sinks) {


Assert the size of this collection so that we know this loop runs.

dlvenable · 2025-08-11T17:35:33Z

I am also thinking that I should rename "FailurePipeline" to "NoSourcePipeline". We may be using this pipeline in other cases as well. @dlvenable what do you think?

I like HeadlessPipeline.

dlvenable · 2025-08-11T17:35:57Z

@kkondaka , The builds are all failing. Please take a look.

Signed-off-by: Kondaka <krishkdk@amazon.com>

dlvenable

Thanks! This will be a significant improvement to how Data Prepper handles errors!

dlvenable reviewed Jun 25, 2025

View reviewed changes

kkondaka force-pushed the dlq-pipeline branch from fd78dc1 to f9aed31 Compare July 14, 2025 15:45

dlvenable requested changes Jul 14, 2025

View reviewed changes

kkondaka force-pushed the dlq-pipeline branch from e56bd77 to e5001d9 Compare July 21, 2025 20:15

kkondaka marked this pull request as ready for review July 21, 2025 21:28

kkondaka requested review from KarstenSchnitter, chenqi0805, dinujoh, engechas, graytaylor0, oeyh, san81, sb2k16 and srikanthjg as code owners July 21, 2025 21:28

graytaylor0 reviewed Jul 28, 2025

View reviewed changes

kkondaka added 6 commits August 11, 2025 08:54

Modified to set failure pipeline in all pipeline components

0352f9c

Signed-off-by: Kondaka <krishkdk@amazon.com>

Fix failing test case

7f0cf95

Signed-off-by: Kondaka <krishkdk@amazon.com>

Modified tests

5656d37

Signed-off-by: Kondaka <krishkdk@amazon.com>

fixed spotless check errors

a4dd794

Signed-off-by: Kondaka <krishkdk@amazon.com>

Addressed comments. Added more tests for 100% code coverage

8a76a05

Signed-off-by: Kondaka <krishkdk@amazon.com>

Fixed javadoc error

6d0bd92

Signed-off-by: Kondaka <krishkdk@amazon.com>

kkondaka force-pushed the dlq-pipeline branch from f134755 to 6d0bd92 Compare August 11, 2025 15:54

Removed unnecessary change to AwsSecretsPluginConfigValueTranslator.java

b9970e0

Signed-off-by: Kondaka <krishkdk@amazon.com>

graytaylor0 reviewed Aug 11, 2025

View reviewed changes

dlvenable reviewed Aug 11, 2025

View reviewed changes

Addressed review comments

fcdb2f3

Signed-off-by: Kondaka <krishkdk@amazon.com>

dlvenable approved these changes Aug 12, 2025

View reviewed changes

graytaylor0 approved these changes Aug 12, 2025

View reviewed changes

kkondaka merged commit 81d5dff into opensearch-project:main Aug 12, 2025
46 of 47 checks passed

kkondaka mentioned this pull request Aug 12, 2025

Provide sinks access to all headless pipelines #5985

Closed


		import org.opensearch.dataprepper.model.source.Source;

		public interface PipelineIf {

Conversation

kkondaka commented Dec 20, 2024

Description

Issues Resolved

Check List

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kkondaka Jul 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kkondaka commented Aug 4, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dlvenable commented Aug 11, 2025

Uh oh!

dlvenable commented Aug 11, 2025

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

kkondaka Jul 11, 2025 •

edited

Loading